package main

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/hex"
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"regexp"
	"runtime"
	"sort"
	"sync"
	"sync/atomic"
	"syscall"
	"time"

	"github.com/labstack/echo"
	"github.com/labstack/echo/middleware"
	"github.com/sirupsen/logrus"

	"github.com/mia0x75/repeater/config"
	"github.com/mia0x75/repeater/hack"
	"github.com/mia0x75/repeater/obscure"
	"github.com/mia0x75/repeater/services"
	"github.com/mia0x75/repeater/util"
)

var (
	lock      *sync.Mutex
	startTime time.Time
	sys_stats map[string]*stats
)

var (
	debugCmd = flag.Bool("debug", false, "enable debug, disabled by default")
)

const HTTP_POST_TIMEOUT = 5

var expConvertTable = regexp.MustCompile("(?i)^ALTER\\sTABLE\\s(`{0,1}(.*?)`{0,1}\\.{0,1})?`{0,1}([^`\\.]+?)`{0,1}\\s(CONVERT\\sTO\\s).*")

var tr = &http.Transport{
	MaxIdleConnsPerHost: 256,
	TLSClientConfig: &tls.Config{
		InsecureSkipVerify: true,
	},
	DisableCompression:  false,
	DisableKeepAlives:   false,
	TLSHandshakeTimeout: 10 * time.Second,
	Dial: func(netw, addr string) (net.Conn, error) {
		deadline := time.Now().Add(HTTP_POST_TIMEOUT * time.Second)
		dial := net.Dialer{
			Timeout:   HTTP_POST_TIMEOUT * time.Second,
			KeepAlive: 86400 * time.Second,
		}
		conn, err := dial.Dial(netw, addr)
		if err != nil {
			return conn, err
		}
		conn.SetDeadline(deadline)
		return conn, nil
	},
}

// timeout
var defaultHttpClient = http.Client{
	Transport: tr,
}

const banner string = `
 ██████╗  ███████╗ ██████╗  ███████╗  █████╗  ████████╗ ███████╗ ██████╗ 
 ██╔══██╗ ██╔════╝ ██╔══██╗ ██╔════╝ ██╔══██╗ ╚══██╔══╝ ██╔════╝ ██╔══██╗
 ██████╔╝ █████╗   ██████╔╝ █████╗   ███████║    ██║    █████╗   ██████╔╝
 ██╔══██╗ ██╔══╝   ██╔═══╝  ██╔══╝   ██╔══██║    ██║    ██╔══╝   ██╔══██╗
 ██║  ██║ ███████╗ ██║      ███████╗ ██║  ██║    ██║    ███████╗ ██║  ██║
 ╚═╝  ╚═╝ ╚══════╝ ╚═╝      ╚══════╝ ╚═╝  ╚═╝    ╚═╝    ╚══════╝ ╚═╝  ╚═╝

`

type stats struct {
	schema  string
	table   string
	inserts int64
	updates int64
	deletes int64
}

func init() {
	time.LoadLocation("Local")
	startTime = time.Now()
	logrus.SetFormatter(&logrus.TextFormatter{
		TimestampFormat:  "2006-01-02 15:04:05",
		ForceColors:      true,
		QuoteEmptyFields: true,
		FullTimestamp:    true})
}

func uptime() time.Duration {
	return time.Since(startTime)
}

func main() {
	fmt.Print(banner)
	logrus.Infof("git commit: %s", hack.Version)
	logrus.Infof("build time: %s", hack.Compile)
	runtime.GOMAXPROCS(runtime.NumCPU())
	flag.Parse()

	// parse configuration file
	path, err := util.GetCurrentPath()
	if err != nil {
		logrus.Errorf("cannot startup project, error: %v", err)
		os.Exit(1)
	}

	cfg, err := config.ParseConfigFile(path + "/config.yaml")
	if err != nil {
		fmt.Printf("parse config file error:%v\n", err.Error())
		os.Exit(1)
		return
	}

	key := cfg.Key

	lock = new(sync.Mutex)
	sys_stats = make(map[string]*stats, 0)

	// initialize web server
	e := echo.New()
	e.HidePort = true
	e.HideBanner = true
	e.Debug = false

	// middleware
	e.Use(middleware.Recover())

	ctx, cancel := context.WithCancel(context.Background())
	service := services.NewHttpService(&ctx, cfg)

	defer service.Close()

	service.Start()

	ob, err := obscure.NewObscure(cfg.Salt, cfg.Key)
	if err != nil {
		logrus.Warnf("%v", err)
		os.Exit(1)
	}

	// routers
	e.GET("/", statistics)
	e.GET("/Help", help)

	e.POST("/Utils/Md5", func(c echo.Context) error {
		if values, err := c.FormParams(); err == nil {
			if plaintext, found := values["plaintext"]; found {
				if len(plaintext) == 1 {
					if len(plaintext[0]) > 0 {
						md5 := ob.Md5(string(plaintext[0]))
						return c.JSON(200, fmt.Sprintf("{\"md5\":\"%s\"}", md5))
					}
				}
			}
		}
		return echo.NewHTTPError(400, "Bad Request")
	})

	e.POST("/Utils/Encrypt", func(c echo.Context) error {
		if values, err := c.FormParams(); err == nil {
			if plaintext, found := values["plaintext"]; found {
				if len(plaintext) == 1 {
					if len(plaintext[0]) > 0 {
						ciphertext := util.Encrypt(util.PKCS7Pad([]byte(plaintext[0])), key)
						return c.JSON(200, fmt.Sprintf("{\"cipher\":\"%s\"}", hex.EncodeToString(ciphertext)))
					}
				}
			}
		}
		return echo.NewHTTPError(400, "Bad Request")
	})

	e.POST("/Utils/Decrypt", func(c echo.Context) error {
		if values, err := c.FormParams(); err == nil {
			if cipher, found := values["cipher"]; found {
				if len(cipher) == 1 {
					if len(cipher[0]) > 0 {
						data, _ := hex.DecodeString(cipher[0])
						plaintext := string(util.PKCS7UPad([]byte(util.Decrypt(data, key))))
						return c.JSON(200, fmt.Sprintf("{\"plaintext\":\"%s\"}", util.Encode(plaintext)))
					}
				}
			}
		}
		return echo.NewHTTPError(400, "Bad Request")
	})

	e.POST("/", func(c echo.Context) error {
		payload := new(config.Payload)      //
		r := c.Request()                    //
		defer r.Body.Close()                //
		body, err := ioutil.ReadAll(r.Body) //
		if err != nil {
			logrus.Warnf("an error occurred when attempt to read data, error: %+v", err)
			return nil
		}
		if *debugCmd {
			fmt.Println(string(body))
		}

		if err := json.Unmarshal([]byte(body), &payload); err != nil {
			logrus.Warnf("invalid request, error: %+v", err)
			return nil
		}
		// 不处理drafts库
		if payload.Database == "drafts" || payload.Database == "mysql" {
			return nil
		}
		var s *stats
		found := false
		name := fmt.Sprintf("%s.%s", payload.Database, payload.Table)

		lock.Lock()
		if s, found = sys_stats[name]; !found {
			s = &stats{
				schema:  payload.Database,
				table:   payload.Table,
				inserts: int64(0),
				updates: int64(0),
				deletes: int64(0),
			}
			sys_stats[name] = s
		}
		lock.Unlock()
		switch payload.EventType {
		case "insert":
			atomic.AddInt64(&s.inserts, int64(1))
		case "update":
			atomic.AddInt64(&s.updates, int64(1))
		case "delete":
			atomic.AddInt64(&s.deletes, int64(1))
		}
		switch payload.EventType {
		case "insert", "update", "delete":
			if err := ob.Handle(payload); err != nil {
				logrus.Warnf("%v", err)
				return nil
			}
			msg, err := prettyJSON(payload)
			if err != nil {
				logrus.Warnf("%v", err)
				return nil
			}
			logrus.Infof("%s", msg)
			service.SendAll(msg)
		case "create", "alter", "drop":
			if payload.Database == "drafts" {
				return nil
			}
			alert(payload, cfg)
		default:
		}
		return c.JSON(200, "{\"msg\" : \"pong\"}")
	})

	// startup
	addr := fmt.Sprintf("%s:%d", cfg.Listen, cfg.Port)
	e.Logger.Fatal(e.Start(addr))

	//
	sc := make(chan os.Signal, 1)
	signal.Notify(sc,
		os.Kill,
		os.Interrupt,
		syscall.SIGHUP,
		syscall.SIGINT,
		syscall.SIGTERM,
		syscall.SIGQUIT)
	<-sc
	cancel()
}

func help(c echo.Context) error {
	help := "binlog repeater, data encryption & decription, hash services\r\n"
	help += "============================================================\r\n"
	help += "1. push binlog to web server end\r\n"
	help += "\r\n"
	help += "2. data encryption\r\n"
	help += "   - path:   /Utils/Encrypt\r\n"
	help += "   - method: POST\r\n"
	help += "   - input:  plaintext=13800138000\r\n"
	help += "   - output: {\"cipher\":\"7f88ed91f1dc19c0b1fffab022be6af8\"}\r\n"
	help += "\r\n"
	help += "3. data decryption\r\n"
	help += "   - path:   /Utils/Decrypt\r\n"
	help += "   - method: POST\r\n"
	help += "   - input:  cipher=7f88ed91f1dc19c0b1fffab022be6af8\r\n"
	help += "   - output: {\"plaintext\":\"13800138000\"}\r\n"
	help += "\r\n"
	help += "4. hash\r\n"
	help += "   - path:   /Utils/Md5\r\n"
	help += "   - method: POST\r\n"
	help += "   - input:  plaintext=13800138000\r\n"
	help += "   - output: {\"md5\":\"e7d341c6253c8c3f0d15517da52b1466\"}\r\n"
	help += "\r\n"
	help += "Author: Michael Xu\r\n"
	help += fmt.Sprintf("git commit: %s\r\n", hack.Version)
	help += fmt.Sprintf("build time: %s\r\n", hack.Compile)
	return c.String(200, help)
}

func statistics(c echo.Context) error {
	c.Response().Writer.Header().Set("content-type", "text/plain")
	c.Response().Writer.Header().Set("refresh", "5; url=./")
	res := fmt.Sprintf("==================+==================================+==============+==============+==============+==============\r\n")
	res += fmt.Sprintf(" %-16s | %-32s | %12s | %12s | %12s | %12s \r\n", "database", "table", "insert(s)", "update(s)", "deletion(s)", "subtotal")
	res += fmt.Sprintf("------------------+----------------------------------+--------------+--------------+--------------+--------------\r\n")
	total := int64(0)
	total_inserts := int64(0)
	total_updates := int64(0)
	total_deletes := int64(0)

	var keys []string
	lock.Lock()
	for name := range sys_stats {
		keys = append(keys, name)
	}
	sort.Strings(keys)
	for _, name := range keys {
		s := sys_stats[name]
		subtotal := s.inserts + s.updates + s.deletes
		total_inserts += s.inserts
		total_updates += s.updates
		total_deletes += s.deletes

		table := s.table
		if len(table) >= 32 {
			table = table[0:29] + "..."
		}
		schema := s.schema
		if len(schema) >= 16 {
			schema = schema[0:13] + "..."
		}
		res += fmt.Sprintf(" %-16s | %-32s | %12d | %12d | %12d | %12d \r\n",
			schema,
			table,
			s.inserts,
			s.updates,
			s.deletes,
			subtotal)
		total += subtotal
	}
	lock.Unlock()
	res += fmt.Sprintf("==================+==================================+==============+==============+==============+==============\r\n")
	res += fmt.Sprintf(" %-51s | %12d | %12d | %12d | %12d \r\n", "grand total", total_inserts, total_updates, total_deletes, total)
	res += fmt.Sprintf("=====================================================+==============+==============+==============+==============\r\n")

	duration := time.Now().Unix() - startTime.Unix()
	changes_rate := float64(total) / float64(duration)
	updates_rate := float64(total_updates) / float64(duration)
	inserts_rate := float64(total_inserts) / float64(duration)
	deletes_rate := float64(total_deletes) / float64(duration)

	res += fmt.Sprintf("\r\n")
	res += fmt.Sprintf("summary\r\n")
	res += fmt.Sprintf("-------------------------------\r\n")
	res += fmt.Sprintf("change(s) per second: %9.2f\r\n", changes_rate)
	res += fmt.Sprintf("insert(s) per second: %9.2f\r\n", inserts_rate)
	res += fmt.Sprintf("update(s) per second: %9.2f\r\n", updates_rate)
	res += fmt.Sprintf("delete(s) per second: %9.2f\r\n", deletes_rate)
	res += "\r\n"
	res += "Author: Michael Xu\r\n"
	res += fmt.Sprintf("git commit: %s\r\n", hack.Version)
	res += fmt.Sprintf("build time: %s\r\n", hack.Compile)

	return c.String(200, fmt.Sprintf("uptime: %v\r\nsystem statistics:\r\n%s", uptime(), res))
}

func alert(p *config.Payload, cfg *config.Config) {
	var buf bytes.Buffer
	query := p.Event.Data["query"].(string)
	if match := expConvertTable.FindSubmatch([]byte(query)); match != nil {
		return
	}

	l := len(cfg.Peoples)
	for _, sub := range cfg.Peoples[:l-1] {
		buf.WriteString(sub.Email)
		buf.WriteString(",")
	}
	buf.WriteString(cfg.Peoples[l-1].Email)

	//
	subject := "attention: database schema changed"
	mailBody := fmt.Sprintf("Database: %s<br>Table: %s<br>Action: %s<br>SQL: %s",
		p.Database,
		p.Table,
		p.EventType,
		query)
	callFrom := "moucx"

	data := make(url.Values)
	data["to"] = []string{buf.String()}
	data["subject"] = []string{subject}
	data["mailBody"] = []string{mailBody}
	data["callFrom"] = []string{callFrom}
	url := "http://172.30.19.121:9081/mailService/sendTextMail"
	res, err := http.PostForm(url, data)
	if err != nil {
		logrus.Warnf("%+v", err)
		return
	}
	defer res.Body.Close()

	body, err := ioutil.ReadAll(res.Body)
	if err != nil {
		logrus.Warnf("%+v", err)
	} else {
		logrus.Infof("%s", string(body))
	}
}

func prettyJSON(data interface{}) ([]byte, error) {
	buffer := new(bytes.Buffer)
	encoder := json.NewEncoder(buffer)
	encoder.SetIndent("", `	`)

	err := encoder.Encode(data)
	return buffer.Bytes(), err
}
